package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/7/10
 * 该案例演示了flatMap算子
 * 需求：从指定的网络端口读取一行数据   并对其进行扁平化处理
 */
public class Flink11_trans_flatMap {
    public static void main(String[] args) throws Exception {
        //TODO 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 从指定网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //注意：如果是map，没有办法向下游传递多条数据，因为它是通过返回值向下游传递数据
        //注意：如果是flaMap，可以向下游传递多条数据，因为它不是通过返回值向下游传递数据，它是通过Collector向下游传递数据
        SingleOutputStreamOperator<String> flatMapDS = socketDS.flatMap(
                new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        for (String word : value.split(" ")) {
                            out.collect(word);
                        }
                    }
                }
        );
        flatMapDS.print();
        //TODO 提交作业
        env.execute();

    }
}